package thread.example.forkjoin;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletionService;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.stream.LongStream;

import org.junit.Test;
import org.springframework.util.StopWatch;

/**
 * @author jiangwz
 * @create 2021/3/12.
 */
public class ForkJoinDemo {

    public static long[] numbers;

    @Test
    public void test01(){
        numbers = LongStream.rangeClosed(1, 10_000_000).toArray();
        long sum = 0;
        for (int i = 0; i < numbers.length; i++) {
            sum += numbers[i];
        }
        System.out.println("sum  = " + sum);
    }

    public static final int THRESHOLD = 10_000;
    private static long allSum;

    @Test
    public void test02() throws InterruptedException {
        numbers = LongStream.rangeClosed(1, 10_000_000).toArray();
        int taskSize = (int) (numbers.length / THRESHOLD);
        for (int i = 1; i <= taskSize; i++) {
            final int key = i;
            new Thread(new Runnable() {
                public void run() {
                    sumAll(sum((key - 1) * THRESHOLD, key * THRESHOLD));
                }
            }).start();
        }
        Thread.sleep(100);
        System.out.println("allSum = " + getAllSum());

    }
    private static synchronized long sumAll(long threadSum) {
        return allSum += threadSum;
    }

    public static synchronized long getAllSum() {
        return allSum;
    }

    private static long sum(int start, int end) {
        long sum = 0;
        for (int i = start; i < end; i++) {
            sum += numbers[i];
        }
        return sum;
    }


    @Test
    public void test03() throws InterruptedException, ExecutionException {
        numbers = LongStream.rangeClosed(1, 10_000_000).toArray();
        ExecutorService executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() + 1);
        CompletionService<Long> completionService = new ExecutorCompletionService<Long>(executor);
        int taskSize = (int) (numbers.length / THRESHOLD);
        for (int i = 1; i <= taskSize; i++) {
            final int key = i;
            completionService.submit(new Callable<Long>() {

                @Override
                public Long call() throws Exception {
                    return sum((key - 1) * THRESHOLD, key * THRESHOLD);
                }
            });
        }
        long sumValue = 0;
        for (int i = 0; i < taskSize; i++) {
            sumValue += completionService.take().get();
        }
        // 所有任务已经完成,关闭线程池
        System.out.println("sumValue = " + sumValue);
        executor.shutdown();

    }



    @Test
    public void test05(){
        int num = 1000_000;
        List<String> list = new ArrayList<>(num);
        for (int i = 0; i < num; i++) {
            if (i%2 ==0) {
                list.add("female");
            }else {
                list.add("male");
            }
        }

        List<String> maleList = new ArrayList<>(num);
        List<String> femaleList = new ArrayList<>(num);
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        for (String s : list) {
            if (s.equals("male")) {
                maleList.add(s);
            }else {
                femaleList.add(s);
            }
        }
        stopWatch.stop();
        System.out.println("cost" + stopWatch.getTotalTimeMillis()); // 527,520,524
        System.out.println("maleList="+maleList.size() + " femaleList="+ femaleList.size());
    }




    volatile List<String> maleLists = new ArrayList<>(1000_00000);
    volatile List<String> femaleLists = new ArrayList<>(1000_00000);

    @Test
    public void test06() throws InterruptedException {
        int num = 1000_000;
        List<String> list = new ArrayList<>(num);
        for (int i = 0; i < num; i++) {
            if (i%2 ==0) {
                list.add("female");
            }else {
                list.add("male");
            }
        }
        StopWatch stopWatch = new StopWatch();
        stopWatch.start();
        int threadNum = 10;
        int threshold = num / threadNum;
        CountDownLatch latch = new CountDownLatch(threadNum);
        ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
        for (int i = 1; i <= threadNum; i++) {
            List<String> stringList = list.subList((i - 1) * threshold, i * threshold);
            System.out.println("size"+stringList.size());
            executorService.submit(() -> {
                concurrentDo(stringList, latch);
            });
        }
        latch.await();
        stopWatch.stop();
        System.out.println("cost" + stopWatch.getTotalTimeMillis());
        System.out.println("maleLists="+maleLists.size() + " femaleLists="+ femaleLists.size());

    }


    private void concurrentDo(List<String> stringList, CountDownLatch latch) {
         List<String> maleList = new ArrayList<>(stringList.size());
         List<String> femaleList = new ArrayList<>(stringList.size());
        for (String s : stringList) {
            if (s.equals("male")) {
                maleList.add(s);
            }else {
                femaleList.add(s);
            }
        }
        maleLists.addAll(maleList);
        femaleLists.addAll(femaleList);
        latch.countDown();

    }


}
